package com.ccb.channel.atm;

import java.io.IOException;
import java.sql.Timestamp;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.log4j.Logger;

import com.alibaba.fastjson.JSON;
import com.ccb.filter.Filters;
import com.ccb.filter.RecordFilter;
import com.ccb.util.CTool;
import com.ccb.util.Record;
import com.ccb.util.RecordType;
import com.ccb.util.XmlJSON;

public class AtmMapper extends Mapper<LongWritable, Text, Text, Text> {
	/**
	 * Logger for this class
	 */
	private static final Logger logger = Logger.getLogger(AtmMapper.class);
	private RecordFilter recordFilter ; 
	@Override
	protected void setup(Context context) throws IOException, InterruptedException {
		recordFilter  =  Filters.getFilter("atm");
		if(recordFilter == null){
			logger.info("fiter is null");
		}
	}
	public Timestamp getAtmReqTime(String year,String time){
		if (logger.isDebugEnabled()) {
			logger.debug("getAtmReqTime(String, String) - start");
		}

		Timestamp returnTime =null;
		String strDate=year+"-"+time.trim();
		try {
			returnTime=Timestamp.valueOf(strDate);
			System.out.println(returnTime);
//			Date d = DateUtil.str2Date("yyyy-MM-dd HH:mm:ss.S", CTool.replace(
//					strDate, "T", " "));
//			strReturn = DateUtil.date2Str("yyyyMMddHHmmssS", d);

		} catch (Exception e) {
			logger.error("getAtmReqTime(String, String)", e);
			
			if (logger.isDebugEnabled()) {
				logger.debug("getAtmReqTime(String, String) - end");
			}
			return returnTime;
		}

		if (logger.isDebugEnabled()) {
			logger.debug("getAtmReqTime(String, String) - end");
		}
		return returnTime;
	}
	
	public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
		if (logger.isDebugEnabled()) {
			logger.debug("map(LongWritable, Text, Context) - start");
		}

		String values = value.toString().trim();
		String recordTypePattern = context.getConfiguration().get("atm.record.recordType.Pattern");
		String year = context.getConfiguration().get("atm.record.year");
		Timestamp returnTime=null;
		String serviceId = "";
		String matchKey = "";
		String rspCode="";
		String reqTime="";
		StringBuffer sbf = new StringBuffer();
		try {
			values = values.replaceAll("\r", "");
			String[] strs = values.split("\n");
			String realValue = "";
			for (String linaValue : strs) {
				reqTime=linaValue.substring(13,31);
				linaValue = linaValue.substring(47);
				returnTime=getAtmReqTime(year,reqTime);
				sbf.append(linaValue);
			}
			realValue = sbf.toString();
			
			serviceId = StringUtils.substringBetween(realValue, "<jydm>", "</jydm>");
			serviceId=CTool.removeCData(serviceId);
			if(recordFilter != null && !recordFilter.accept(serviceId)){
				return;
			}
			
			matchKey = StringUtils.substringBetween(realValue, "<sbbh>", "</sbbh>");
			matchKey=CTool.removeCData(matchKey);
			if (serviceId == null || matchKey == null)return;
			
				
			Record rd = new Record();
			rd.setServiceId(serviceId);
			rd.setUid(matchKey);
			rd.setOriginalData(realValue);
			rd.setData(XmlJSON.xmlToJSONObj(realValue));
			rd.setTime(returnTime);
			if (!realValue.matches(recordTypePattern)) {
				rd.setRecordType(RecordType.REQUEST);
			} else {
				rd.setRecordType(RecordType.RESPONSE);
				rspCode=StringUtils.substringBetween(realValue, "<xym>", "</xym>");
				rd.setResponseCode(rspCode);
			}
			context.write(new Text(serviceId), new Text(JSON.toJSONString(rd)));
		} catch (Exception e) {
			logger.error("map(LongWritable, Text, Context)", e);

			if (logger.isDebugEnabled()) {
				logger.debug("map(LongWritable, Text, Context) - end");
			}
			return;
		}

		if (logger.isDebugEnabled()) {
			logger.debug("map(LongWritable, Text, Context) - end");
		}
	}
}
